//package kafka
//
//import org.apache.kafka.clients.consumer.ConsumerRecord
//import org.apache.kafka.common.serialization.StringDeserializer
//import org.apache.spark.SparkConf
//import org.apache.spark.streaming.dstream.{DStream, InputDStream}
//import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
//import org.apache.spark.streaming.kafka010.KafkaUtils
//import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
//import org.apache.spark.streaming.{Seconds, StreamingContext}
//import org.sparkproject.guava.eventbus.Subscribe
//
//object KafkaStreaming01 {
//  def main(args: Array[String]): Unit = {
//    //创建sparkConf
//    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("ks01")
//    //创建StreamingContext，时间间隔为1s（采集
//    val ssc = new StreamingContext(sparkConf,Seconds(1))
//    /*
//      *配置kafka消费者参数：
//      * bootstrap.servers  kafka的服务节点和端口号，用于连接kafka
//      * key.deserializer   网络传输要序列化，接收要反序列化，这里是反序列化的类
//      * value.deserializer 指定反序列化的类
//      * group.id           消费者组ID，ID相同的消费者在同一个组
//      * enable.auto.commit kafka是否自动提交偏移量，这里填否，交由Spark管理
//     */
//    val kafkaParams = Map[String,Object](
//      "bootstrap.servers" -> "node01:9092,node02:9092,node03:9092",
//      "key.deserializer" -> classOf[StringDeserializer],
//      "group.id" -> "wordcount",
//      "enable.auto.commit" -> (false:java.lang.Boolean)
//    )
//    //定义一个主题数组，内可包含多个主题，此处只有一个
//    val topics = Array("wordcount")
//    //创建Kafka数据源
//    val linesDStream:InputDStream[ConsumerRecord[String,String]] =
//      KafkaUtils.createDirectStream(ssc,PreferConsistent,
//        Subscribe[String,String](topics,kafkaParams))
//    //取出一行，只要其中的value值
//    val lineRDD:DStream[String] = linesDStream.map(_.value())
//    //分离单词
//    val wordRDD:DStream[String] = lineRDD.flatMap(_.split("\\s+"))
//    //转换（单词，1）
//    val wordAndOneRDD:DStream[(String,Int)] = wordRDD.map((_,1))
//    //统计
//    val resultRDD:DStream[(String,Int)] = wordAndOneRDD.reduceByKey(_+_)
//    //foreachRDD(func)最通用的输出操作，把func作用于从stream生成的每一个RDD
//    resultRDD.foreachRDD(rdd=>rdd.foreach(println))
//    ssc.start()
//    ssc.awaitTermination()
//  }
//
//}
